package com.sui.bigdata.flink.sql.function.aggregate;

import org.apache.flink.table.functions.AggregateFunction;
import com.sui.bigdata.flink.sql.function.util.UserLoanCntObject;

import java.util.Iterator;

/**
 * @author dongguosheng
 * @date 2019/12/11 11:19
 * @description
 * @reviewer
 */
public class GangLoanUserCnt extends AggregateFunction<Integer, UserLoanCntObject> {
    @Override
    public Integer getValue(UserLoanCntObject userLoanCntObject) {
        return userLoanCntObject.getLoanCnt();
    }

    public void accumulate(UserLoanCntObject userLoanCntObject, String graphFraudGrade, Integer graphSize, Integer loanCnt) {
        if (graphFraudGrade != null && !"".equals(graphFraudGrade)) {
            char graphFraudGradenew = graphFraudGrade.charAt(0);
            if (userLoanCntObject.getGraphFraudGrade() < graphFraudGradenew) {
                userLoanCntObject.setGraphSize(graphSize);
                userLoanCntObject.setLoanCnt(loanCnt);
                userLoanCntObject.setGraphFraudGrade(graphFraudGradenew);
            } else if (userLoanCntObject.getGraphFraudGrade() == graphFraudGradenew) {
                if (userLoanCntObject.getGraphSize() < graphSize) {
                    userLoanCntObject.setGraphSize(graphSize);
                    userLoanCntObject.setLoanCnt(loanCnt);
                    userLoanCntObject.setGraphFraudGrade(graphFraudGradenew);
                } else if (userLoanCntObject.getGraphSize() == graphSize) {
                    if (userLoanCntObject.getLoanCnt() < loanCnt) {
                        userLoanCntObject.setGraphSize(graphSize);
                        userLoanCntObject.setLoanCnt(loanCnt);
                        userLoanCntObject.setGraphFraudGrade(graphFraudGradenew);
                    }
                }
            }
        } else {
            if (graphSize != null){
                if (userLoanCntObject.getGraphSize() < graphSize) {
                    userLoanCntObject.setGraphSize(graphSize);
                    userLoanCntObject.setLoanCnt(loanCnt);
                    userLoanCntObject.setGraphFraudGrade('0');
                } else if (userLoanCntObject.getGraphSize() == graphSize) {
                    if (userLoanCntObject.getLoanCnt() < loanCnt) {
                        userLoanCntObject.setGraphSize(graphSize);
                        userLoanCntObject.setLoanCnt(loanCnt);
                        userLoanCntObject.setGraphFraudGrade('0');
                    }
                }
            }else {
                if (userLoanCntObject.getLoanCnt() < loanCnt) {
                    userLoanCntObject.setGraphSize(graphSize);
                    userLoanCntObject.setLoanCnt(loanCnt);
                    userLoanCntObject.setGraphFraudGrade('0');
                }
            }

        }
    }

    public void merge(UserLoanCntObject userLoanCntObject, Iterable<UserLoanCntObject> it) {
        Iterator<UserLoanCntObject> iter = it.iterator();
        while (iter.hasNext()) {
            UserLoanCntObject a = iter.next();
            if (userLoanCntObject.getGraphFraudGrade() < a.getGraphFraudGrade()) {
                userLoanCntObject.setGraphSize(a.getGraphSize());
                userLoanCntObject.setLoanCnt(a.getLoanCnt());
                userLoanCntObject.setGraphFraudGrade(a.getGraphFraudGrade());
            } else if (userLoanCntObject.getGraphFraudGrade() == a.getGraphFraudGrade()) {
                if (userLoanCntObject.getGraphSize() < a.getGraphSize()) {
                    userLoanCntObject.setGraphSize(a.getGraphSize());
                    userLoanCntObject.setLoanCnt(a.getLoanCnt());
                    userLoanCntObject.setGraphFraudGrade(a.getGraphFraudGrade());
                } else if (userLoanCntObject.getGraphSize() == a.getGraphSize()) {
                    if (userLoanCntObject.getLoanCnt() < a.getLoanCnt()) {
                        userLoanCntObject.setGraphSize(a.getGraphSize());
                        userLoanCntObject.setLoanCnt(a.getLoanCnt());
                        userLoanCntObject.setGraphFraudGrade(a.getGraphFraudGrade());
                    }
                }
            }
        }
    }

    @Override
    public UserLoanCntObject createAccumulator() {
        return new UserLoanCntObject();
    }
}
